package com.boboPlanet.user.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.boboPlanet.comm.context.BusinessThreadContextHolder;
import com.boboPlanet.comm.entity.dto.PlanetMqMsg;
import com.boboPlanet.comm.enums.PlanetCommCode;
import com.boboPlanet.comm.exception.BusinessException;
import com.boboPlanet.comm.rabbitmq.RabbitMqRoutingConfig;
import com.boboPlanet.user.async.AsyncTask;
import com.boboPlanet.user.constants.ElasticDslConstants;
import com.boboPlanet.user.entity.dto.CreateArtEsIndexDTO;
import com.boboPlanet.user.entity.dto.SysLogDTO;
import com.boboPlanet.user.entity.po.ArticleDetailPO;
import com.boboPlanet.user.entity.po.ArticleMainInfoPO;
import com.boboPlanet.user.enums.PlanetUserEsIndexEnum;
import com.boboPlanet.user.enums.SysOperatorEnum;
import com.boboPlanet.user.mapper.ArticleDetailMapper;
import com.boboPlanet.user.mapper.ArticleMapper;
import com.boboPlanet.user.single.SingleFactory;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectInputStream;

/**
 * @Description: TODO ES 保存星球数据 MQ
 * @Author: mis_wu
 * @Date: 2023/6/21
 * @Company planet
 * 也可使用ElasticsearchTemplate \ ElasticsearchOperations 操作ES
 **/
@Slf4j
@Component
public class ElasticInsertPlanetDocConsume {

    @Resource
    private RestHighLevelClient restHighLevelClient;

//    @Resource
//    private ElasticsearchOperations elasticsearchTemplate;


    @Resource
    private ArticleMapper articleMapper;

    @Resource
    private ArticleDetailMapper articleDetailMapper;

    @Resource
    private AsyncTask asyncTask;


//    @RabbitListener(bindings = {@QueueBinding(
//            value = @Queue("esInsertQueue"),
//            exchange = @Exchange(value = "esInsertExchange",type = ExchangeTypes.TOPIC),
//            key = {"es-insert-article"}
//    )})
    @RabbitListener(queues = RabbitMqRoutingConfig.ES_INSERT_ARTICLE,ackMode = "MANUAL")
    public void insertArticleToEs(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {

        String dataStr = new String(message.getBody());

        PlanetMqMsg mqData = PlanetMqMsg.deserialize(dataStr);

        log.info("receive sync es data rabbitMq success,data is: {}",mqData);
        //todo:先判断是否存在 es index
        GetIndexRequest request = new GetIndexRequest(PlanetUserEsIndexEnum.PLANET_ARTICLE_INFO_INDEX.getName());
        try {
            boolean exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            if (!exists){
                this.createIndex(PlanetUserEsIndexEnum.PLANET_ARTICLE_INFO_INDEX.getName());
            }
        }catch (Exception e){
            log.error("insertArticleToEs MQ Exception,create es index fail,reason: {}",e.getMessage(),e);
            throw BusinessException.build("创建es索引失败,请联系管理员");
        }

        ArticleMainInfoPO mainPO = articleMapper.queryArtMainById((Long) mqData.getData());
        if (ObjectUtils.isEmpty(mainPO)){
            log.error("insert into es doc,article main info not exist,articleId is: {}", mqData.getData());
            return;
        }
        ArticleDetailPO detailPO = articleDetailMapper.queryDetailById((Long) mqData.getData());

        CreateArtEsIndexDTO dto = new CreateArtEsIndexDTO();
        dto.setArticleId((Long) mqData.getData());
        dto.setUserId(mainPO.getUserId());
        dto.setArticleUser(mainPO.getArticleUser());
        dto.setAccountId(mainPO.getAccountId());
        dto.setArticleTypeId(mainPO.getArticleTypeId());
        dto.setArticleTypeName(mainPO.getArticleTypeName());
        dto.setArticleTitle(mainPO.getArticleTitle());
        dto.setIsAnonymous(mainPO.getIsAnonymous());
        dto.setPlanetUpdateTime(mainPO.getPlanetUpdateTime());
        dto.setPlanetUpdateUser(mainPO.getPlanetUpdateUser());
        dto.setArticleState(mainPO.getArticleState());
        dto.setVersion(mainPO.getVersion());
        dto.setArticleDetailId(detailPO.getArticleDetailId());
        dto.setArticleContent(detailPO.getArticleContent());

        //todo: 存在则插入数据
        IndexRequest idxRequest = new IndexRequest(PlanetUserEsIndexEnum.PLANET_ARTICLE_INFO_INDEX.getName()).id((String) mqData.getData());

        idxRequest.source(JSON.toJSONString(dto),XContentType.JSON);

        try {
            restHighLevelClient.index(idxRequest,RequestOptions.DEFAULT);

            //手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

        } catch (Exception e) {
            // 手动拒绝消息 args[MQTag, 是否批量处理, 是否重新放入队列]
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            log.error("rabbitmq consume fail,reason: {}",e.getMessage(),e);
        }

        SysLogDTO logDto = SingleFactory.getLogDto(SysOperatorEnum.ES_INSERT_ART_DOC.getDesc());

        asyncTask.tranceLog(logDto);

    }

    /**
     * 创建索引库
     * @param indexName 索引名称
     */
    public void createIndex(String indexName) throws IOException {

        CreateIndexRequest request = new CreateIndexRequest(indexName);

        request.source(ElasticDslConstants.CREATE_ARTICLE_INFO_INDEX_DSL, XContentType.JSON);

        restHighLevelClient.indices().create(request,RequestOptions.DEFAULT);

        SysLogDTO logDto = SingleFactory.getLogDto(SysOperatorEnum.ES_CREATE_ART_IDX.getDesc());

        asyncTask.tranceLog(logDto);
    }
}
